// @Author : zaniu(zzaniu@126.com)
// @Time   : 2021/9/1 17:17
// @Desc   :
//
//                   _ooOoo_
//                  o8888888o
//                  88" . "88
//                  (| -_- |)
//                  O\  =  /O
//               ____/`---'\____
//             .'  \\|     |//  `.
//            /  \\|||  :  |||//  \
//           /  _||||| -:- |||||-  \
//           |   | \\\  -  /// |   |
//           | \_|  ''\---/''  |   |
//           \  .-\__  `-`  ___/-. /
//         ___`. .'  /--.--\  `. . __
//      ."" '<  `.___\_<|>_/___.'  >'"".
//     | | :  `- \`.;`\ _ /`;.`/ - ` : | |
//     \  \ `-.   \_ __\ /__ _/   .-` /  /
// =====`-.____`-.___\_____/___.-`____.-'=====
//                   `=---='
// ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
//           佛祖保佑       永无BUG
//
package main

import (
	"context"
	"distributedTransaction/proto2go"
	"distributedTransaction/utils"
	"distributedTransaction/utils/nacos"
	"encoding/json"
	"fmt"
	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/consumer"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"log"
	"net"
	"os"
	"os/signal"
	"unsafe"
)

type InventoryServer struct{}

var sql *utils.Sql

func (inv InventoryServer) QueryInventoryRecord(ctx context.Context, Request *proto2go.QueryInventoryRequest) (*proto2go.QueryInventoryReply, error) {
	existsFlag := false
	// ret := sql.RawSelect("select count(*) COUNT from inventory where SerialNo = ?", Request.GlobalNo)
	ret := sql.Select("inventory_record", "id", map[string]interface{}{"GlobalNo": Request.GlobalNo}, 1)
	if len(*ret) != 0 {
		existsFlag = true
		fmt.Println("已查询到库存记录, 流水号： ", Request.GlobalNo)
	} else {
		fmt.Println("未查询到库存记录, 流水号： ", Request.GlobalNo)
	}
	return &proto2go.QueryInventoryReply{Exists: existsFlag}, nil
}

func (inv InventoryServer) DetainInventory(ctx context.Context, Request *proto2go.InventoryRequest) (*proto2go.InventoryReply, error) {
	retFlag := false
	tx := sql.Begin()
	n := tx.RawSql("Update inventory set GoodsNumber=GoodsNumber-? where GoodsName=? and GoodsNumber > ?", Request.GoodsNumber, Request.GoodsName, Request.GoodsNumber)
	if n > 0 {
		ret := tx.Select("inventory", "id", map[string]interface{}{"GoodsName": Request.GoodsName}, 1)
		n2 := tx.Insert("inventory_record", []string{"GlobalNo", "UserName", "GoodsName", "GoodsNumber", "InventoryId"}, []interface{}{Request.GlobalNo, Request.UserName, Request.GoodsName, Request.GoodsNumber, (*ret)[0]["id"]})
		if n2 > 0 {
			retFlag = true
			tx.Commit()
			fmt.Printf("扣减库存成功, 流水号： %s, 用户： %s, 商品： %s, 数量: %d\n", Request.GlobalNo, Request.UserName, Request.GoodsName, Request.GoodsNumber)
		} else {
			tx.Rollback()
			fmt.Printf("扣减库存失败, 请稍后再试, 流水号： %s, 用户： %s, 商品： %s, 数量: %d\n", Request.GlobalNo, Request.UserName, Request.GoodsName, Request.GoodsNumber)
			return &proto2go.InventoryReply{Flag: retFlag}, status.Errorf(codes.NotFound, "库存扣减失败")
		}
	} else {
		tx.Rollback()
		fmt.Printf("扣减库存失败, 流水号： %s, 用户： %s, 商品： %s, 数量: %d\n", Request.GlobalNo, Request.UserName, Request.GoodsName, Request.GoodsNumber)
		return &proto2go.InventoryReply{Flag: retFlag}, status.Errorf(codes.NotFound, "库存扣减失败")
	}
	return &proto2go.InventoryReply{Flag: retFlag}, nil
}

func (inv InventoryServer) RollbackInventory(ctx context.Context, Request *proto2go.InventoryRequest) (*proto2go.InventoryReply, error) {
	retFlag := false
	tx := sql.Begin()
	n := tx.RawSql("Update inventory set GoodsNumber=GoodsNumber+? where GoodsName=GoodsName", Request.GoodsName, Request.GoodsNumber)
	if n > 0 {
		n2 := tx.Delete("inventory_record", map[string]interface{}{"GlobalNo": Request.GlobalNo})
		if n2 > 0 {
			retFlag = true
			tx.Commit()
			fmt.Printf("归还库存成功, 流水号： %s, 用户： %s, 商品： %s, 数量: %d\n", Request.GlobalNo, Request.UserName, Request.GoodsName, Request.GoodsNumber)
		} else {
			fmt.Printf("归还库存失败, 请稍后再试, 流水号： %s, 用户： %s, 商品： %s, 数量: %d\n", Request.GlobalNo, Request.UserName, Request.GoodsName, Request.GoodsNumber)
			tx.Rollback()
		}
	} else {
		tx.Rollback()
		fmt.Printf("归还库存失败, 流水号： %s, 用户： %s, 商品： %s, 数量: %d\n", Request.GlobalNo, Request.UserName, Request.GoodsName, Request.GoodsNumber)
	}
	return &proto2go.InventoryReply{Flag: retFlag}, nil
}

func rocketMqClient() error {
	c, _ := rocketmq.NewPushConsumer(
		consumer.WithGroupName("testGroup5"),
		consumer.WithNsResolver(primitive.NewPassthroughResolver([]string{fmt.Sprintf("%s:%d", utils.GlobalServerConfig.RocketMQHost, utils.GlobalServerConfig.RocketMQPort)})),
		consumer.WithConsumeFromWhere(consumer.ConsumeFromLastOffset), // 消费最后一个消息
		// consumer.WithConsumerModel(consumer.BroadCasting),  // 非广播, 默认就是非广播
	)
	err := c.Subscribe(utils.GlobalServerConfig.Topic, consumer.MessageSelector{}, func(ctx context.Context,
		msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
		msgInfo := utils.MsgInfo{}
		err := json.Unmarshal(msgs[0].Body, &msgInfo)
		if err != nil {
			fmt.Println("消息格式异常，请联系管理员， msg = ", *(*string)(unsafe.Pointer(&msgs[0].Body)))
			return consumer.ConsumeSuccess, err
		}
		// 这里应该是去查数据库，查到就调用回滚库存的，查不到就不处理
		tx := sql.Begin()
		ret := tx.Select("inventory_record", "id", map[string]interface{}{"GlobalNo": msgInfo.GlobalNo}, 1)
		if len(*ret) > 0 {
			n1 := tx.Delete("inventory_record", map[string]interface{}{"GlobalNo": msgInfo.GlobalNo})
			n2 := tx.RawSql("Update inventory set GoodsNumber=GoodsNumber+? where GoodsName=?", msgInfo.GoodsNumber, msgInfo.GoodsName)
			if n1 > 0 && n2 > 0 {
				tx.Commit()
				fmt.Printf("归还库存成功, 流水号： %s, 用户： %s, 商品： %s, 数量: %d\n", msgInfo.GlobalNo, msgInfo.UserName, msgInfo.GoodsName, msgInfo.GoodsNumber)
				return consumer.ConsumeSuccess, nil
			} else {
				tx.Rollback()
				fmt.Printf("归还库存失败, 稍后重试, 流水号： %s, 用户： %s, 商品： %s, 数量: %d\n", msgInfo.GlobalNo, msgInfo.UserName, msgInfo.GoodsName, msgInfo.GoodsNumber)
				return consumer.ConsumeRetryLater, nil
			}
		} else {
			// 这里是sql库没有设计好， 应该要返回err的。更新返回0，说明没找到库存
			tx.Rollback()
			fmt.Printf("归还库存失败， 库存记录不存在, 流水号： %s, 用户： %s, 商品： %s, 数量: %d\n", msgInfo.GlobalNo, msgInfo.UserName, msgInfo.GoodsName, msgInfo.GoodsNumber)
			return consumer.ConsumeSuccess, nil
		}
	})
	if err != nil {
		return err
	}
	// Note: start after subscribe
	err = c.Start()
	return err
}

func register(ip string, port uint64, serverName string) {
	ok, err := nacos.Register(ip, port, serverName)
	if err != nil || !ok {
		fmt.Println("注册失败, err = ", err.Error())
		os.Exit(-1)
	}
	fmt.Println("注册成功")
}

func deregister(ip string, port uint64, serverName string) {
	nacos.Deregister(ip, port, serverName)
}

func main() {
	ip := "127.0.0.1"
	port := 8080
	serverName := "gmb.go.inventory"
	nacos.ListenServerConfig("dev-data", "dev-group", &utils.GlobalServerConfig)
	sql = utils.NewSql()
	register(ip, uint64(port), serverName)
	err := rocketMqClient()
	if err != nil {
		fmt.Println("err = ", err.Error())
		os.Exit(-1)
	}
	inventoryServer := grpc.NewServer()
	// 将服务注册到 gRPC server 实例
	proto2go.RegisterInventoryServer(inventoryServer, InventoryServer{})
	// 监听 8080 端口
	listen, err := net.Listen("tcp", fmt.Sprintf("%s:%d", ip, port))
	defer listen.Close()
	if err != nil {
		panic("net.Listen err: " + err.Error())
	}
	go func() {
		// 运行服务
		if err := inventoryServer.Serve(listen); err != nil {
			panic("server.Serve err: " + err.Error())
		}
	}()
	quit := make(chan os.Signal)
	signal.Notify(quit, os.Interrupt)
	<-quit
	nacos.CancelListenConfig("dev-data", "dev-group")
	deregister(ip, uint64(port), serverName)
	inventoryServer.GracefulStop()
	log.Println("服务退出 ...")
}
